package chapter_04

import org.apache.spark.{SparkConf, SparkContext}


class chapter_s04 {
  val conf = new SparkConf().setMaster("local").setAppName("My App")
  val sc = new SparkContext(conf)
  val data = Set((1, 2), (3, 4), (3, 6))
  val data2 = Set((3, 9))
  val data3 = Set(("panda", 0), ("pink", 3), ("pirate", 3), ("panda", 1),
    ("pink", 4))
  val rdd = sc.parallelize(data.toList)
  val rdd2 = sc.parallelize(data2.toList)
  val rdd3 = sc.parallelize(data3.toList)

  // 单个Pair RDD的转换操作
  ///////////////////////

  def reduceByKey(): Unit = {
    // 合并具有相同键的值
    val res = rdd.reduceByKey((x, y) => x + y)
    res.foreach(x => print(x.toString + " "))
    // (1,2) (3,10)
  }

  def groupByKey(): Unit = {
    // 对具有相同键的值进行分组
    val res = rdd.groupByKey()
    res.foreach(x => print(x.toString + " "))
    // (1,CompactBuffer(2)) (3,CompactBuffer(4, 6))
  }

  def combineByKey(): Unit = {
    // 使用combineByKey()求每个键对应的平均值
    val res = rdd.combineByKey(
      (v) => (v, 1),
      (acc: (Int, Int), v) => (acc._1 + v, acc._2 + 1),
      (acc1: (Int, Int), acc2: (Int, Int)) => (acc1._1 + acc2._1, acc1._2 + acc2._2)
    ).map{ case (key, value) => (key, value._1 / value._2.toFloat) }
    res.collectAsMap().map(println)
  }

  def mapValues(): Unit = {
    // 对pair RDD中的每个值应用一个函数而不改变键
    val res = rdd.mapValues(x => x + 1)
    res.foreach(x => print(x.toString + " "))
    // (1,3) (3,5) (3,7)
  }

  def flatMapValues(): Unit = {
    // 对pair RDD中的每个值应用一个返回迭代器的函数，然后
    // 对返回的每个元素都生成一个对应原键的键值对记录。
    // 通常用于符号化。
    val res = rdd.flatMapValues(x => (x to 5))
    res.foreach(x => print(x.toString() + " "))
    // (1,2) (1,3) (1,4) (1,5) (3,4) (3,5)
  }

  def keys(): Unit = {
    // 返回一个仅包含键的RDD
    rdd.keys.foreach(x => print(x.toString + " "))
    // 1 3 3
  }

  def values(): Unit = {
    // 返回一个仅包含值的RDD
    rdd.values.foreach(x => print(x.toString + " "))
    // 2 4 6
  }

  def sortByKey(): Unit = {
    // 返回一个根据键排序的RDD
    rdd.sortByKey().foreach(x => print(x.toString + " "))
    // (1,2) (3,4) (3,6)
  }


  // 两个Pair RDD的转换操作
  ///////////////////////

  def subtractByKey(): Unit = {
    // 删除RDD中键与other RDD中的键相同的元素
    rdd.subtractByKey(rdd2).foreach(x => print(x.toString + " "))
  }

  def join(): Unit = {
    // 对两个RDD进行内连接
    rdd.join(rdd2).foreach(x => print(x.toString + " "))
  }

  def rightOuterJoin(): Unit = {
    // 对两个RDD进行连接操作，确保第一个RDD的键必须存在（右外连接）
    rdd.rightOuterJoin(rdd2).foreach(x => print(x.toString + " "))
  }

  def leftOuterJoin(): Unit = {
    // 对两个RDD进行连接操作，确保第二个RDD的键必须存在（左外连接）
    rdd.leftOuterJoin(rdd2).foreach(x => print(x.toString + " "))
  }

  def cogroup(): Unit = {
    // 将两个RDD中拥有相同键的数据分组到一起
    rdd.cogroup(rdd2).foreach(x => print(x.toString + " "))
  }

  // 其他的测试
  def calcAverage(): Unit = {
    // 计算每个键对应的平均值
    rdd3.mapValues(x => (x.toFloat, 1)).
      reduceByKey((x, y) => (x._1 + y._1, x._2 + y._2)).
      mapValues(x => x._1 / x._2).foreach(println)
  }

  def countWords() = {
    // 单词计数
    val input = sc.textFile("pom.xml")
    val words = input.flatMap(x => x.split(" "))
    words.map(x => (x, 1)).reduceByKey((x, y) => x + y).foreach(println)
//    words.countByValue()  更快地实现单词计数
  }
}


object chapter_s04 {

  def main(args: Array[String]): Unit = {
    val o = new chapter_s04
//    o.reduceByKey()
//    o.groupByKey()
//    o.mapValues()
//    o.flatMapValues()
//    o.keys()
//    o.values()
//    o.sortByKey()

    // 两个Pair RDD的转化操作
//    o.subtractByKey()
//    o.join()
//    o.rightOuterJoin()
//    o.leftOuterJoin()
//    o.cogroup()

    // 其他
//    o.calcAverage()
    o.countWords()
  }
}
